#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import locale
import os
import shlex
import socket
import subprocess
import time
from datetime import datetime, timedelta, timezone

from common.cleaner import clear
from common.common import execute_cmd_list, execute_cmd, invoke_rpc_tool_interface
from common.const import AuthType, CMDResult
from sqlserver import log
from sqlserver.commons.common import get_env_variable, check_del_dir_or_file, retry_when_return_false
from sqlserver.commons.const import ExecCmdResult, SQLServerStrConstant, TmpFile, CodeIdConst, SqlServerDefaultUser, \
    SqlCmd


def get_win_dir():
    """
    功能描述: 适配windows下特殊场景路径
    参数：无
    """
    head_path = os.environ['SystemRoot']
    if head_path:
        return head_path.replace("\\", "/")
    return ""


def get_powershell_path():
    """
    功能描述: 获取powershell工具路径
    参数：无
    """
    return f"{get_win_dir()}/system32/WindowsPowerShell/v1.0/powershell"


def check_powershell_result(ps_command, expect_word):
    ps_loc = get_powershell_path()
    process = subprocess.Popen([ps_loc, ps_command], stdout=subprocess.PIPE)
    # 实时获取输出
    while True:
        output = process.stdout.readline()
        if output == b'' and process.poll() is not None:
            break
        if output:
            output_str = output.strip().decode('utf-8')
            if expect_word in output_str:
                return True
    return False


def wait_service_status(service_name, expect_status):
    for _ in range(10):
        if check_powershell_result(f"Get-Service -Name '{service_name}'| Select-Object Status", f"{expect_status}"):
            log.info(f"service name {service_name} stopped successfully")
            return True
        log.debug(f"service name {service_name} {expect_status} continue")
        time.sleep(6)
    log.error(f"service name {service_name} {expect_status} failed")
    return False


def get_service_name(instance_name):
    if instance_name == SQLServerStrConstant.DEFAULT_INSTANCE_NAME:
        return "mssqlserver"
    else:
        return f"MSSQL${instance_name}"


@retry_when_return_false(retry_times=3, delay=3)
def stop_mssql_instance(instance_name):
    service_name = get_service_name(instance_name)
    exec_cmd = f"net stop {service_name} /y"
    ret, std_out, std_err = execute_cmd(exec_cmd)
    log.info(f"ret:{ret},std_out:{std_out},std_err:{std_err},cmd:{exec_cmd}")
    return wait_service_status(service_name, "Stopped")


@retry_when_return_false(retry_times=3, delay=3)
def start_mssql_instance(instance_name, auth_mode, user_name, single_mode: bool = False):
    service_name = get_service_name(instance_name)
    exec_cmd = f"net start {service_name}"
    if single_mode:
        exec_cmd = f"{exec_cmd} /m"
    if str(auth_mode) == str(AuthType.OS_PASSWORD.value):
        exec_cmd = f'{exec_cmd} \"system\"'
    elif str(auth_mode) == str(AuthType.APP_PASSWORD.value):
        exec_cmd = f'{exec_cmd} \"{user_name}\"'
    ret, std_out, std_err = execute_cmd(exec_cmd)
    log.info(f"ret:{ret},std_out:{std_out},std_err:{std_err},cmd:{exec_cmd}")
    return wait_service_status(service_name, "Running")


def get_offset_to_utc():
    utc_offset = datetime.now(tz=timezone.utc).astimezone().utcoffset()
    offset_hours = utc_offset.total_seconds() / 3600
    log.info(f"offet hours: {offset_hours}")
    return offset_hours


def get_offset_to_utc_from_copy(copies):
    if not copies:
        return False, 0
    copy = copies[0]
    if not copy.get("extendInfo", {}).get("offset_to_utc"):
        return False, 0
    return True, copy.get("extendInfo", {}).get("offset_to_utc")


def get_format_restore_time(restore_time, copies):
    local_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(restore_time)))
    utc_time = datetime.utcfromtimestamp(int(restore_time))
    result, offset = get_offset_to_utc_from_copy(copies)
    if result:
        get_origin_backup_time = utc_time + timedelta(hours=float(offset))
        restore_time = get_origin_backup_time.strftime('%Y-%m-%d %H:%M:%S')
        return restore_time
    return local_time


class SqlServerUtils:
    def __init__(self, auth_mode, user_info, pass_prefix, job_id):
        self.auth_mode = auth_mode
        self.user_info = user_info
        self.pass_prefix = pass_prefix
        self.job_id = job_id

    @staticmethod
    def get_replica_id(replica_info):
        # 获取主副本ID
        replica_id = None
        for info in replica_info:
            is_local = info.split(" ")[1]
            role = info.split(" ")[2]
            if is_local == "1" and role == "1":
                replica_id = info.split(" ")[0]
                break
        return replica_id

    @staticmethod
    def list_to_str(in_list):
        result_str = ""
        for mbr in in_list:
            result_str += mbr
            result_str += ","
        if result_str:
            return result_str.strip(',')
        return result_str

    @staticmethod
    def check_service_is_running(service_name):
        """
        # 检查windows下服务是否正常运行，运行中STATE为RUNNING，停止状态该值为STOPPED：
        例如：执行sc query MSSQL$MSSQLSERVER_142 | findstr STATE，查询结果为：
                STATE              : 4  RUNNING
                sc query MSSQLSERVER | findstr STATE
        """
        if service_name.upper() != SQLServerStrConstant.SQLSERVER_DEFAULT_SERVICES.upper():
            service_name = f"MSSQL${service_name}"
        sc_query = "sc query %s " % service_name
        cmd_list = [sc_query, " findstr STATE"]
        ret, std_out, std_err = execute_cmd_list(cmd_list)
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get state of {service_name}, err: {std_err}, out: {std_out}")
            return False
        for out in std_out.split('\n'):
            if "RUNNING" in out:
                log.info(f"The {service_name} is running.")
                return True
        log.error(f"The {service_name} is not running , err: {std_err}, out: {std_out}")
        return False

    @staticmethod
    def read_tmp_result_file(file_path):
        with open(file_path, "r", encoding='UTF-8') as out_file:
            ret = out_file.read()
        check_del_dir_or_file(file_path)
        ret = ret.encode('utf-8').decode("utf-8-sig")
        if len(ret.split("\n")) < 3:
            log.error(f"Failed to read file path: {file_path}")
            raise Exception("Failed to read tmp result")
        return ret

    @staticmethod
    def get_hostname():
        return socket.gethostname().strip()

    @staticmethod
    def read_tmp_result_file_by_line(file_path):
        tmp_code = locale.getdefaultlocale()[1]
        with open(file_path, "r", encoding=tmp_code) as f_handle:
            ret = f_handle.readlines()
        check_del_dir_or_file(file_path)
        return ret

    @staticmethod
    def get_str_by_one_spcase(in_str):
        """
        获取最初的字符串，包含1个空格
        :param in_str: "test test2       test3"
        :return: "test test2"
        """
        indexs = 0
        pre_flags = False
        for str_mbr in in_str:
            if str_mbr == ' ' and pre_flags:
                break
            if str_mbr == ' ':
                pre_flags = True
            else:
                pre_flags = False
            indexs += 1
        return in_str[:indexs].strip()

    @staticmethod
    def find_user_in_list(user_name, user_list, flags):
        if flags:
            return user_name in user_list
        for mbr in user_list:
            if mbr.find(user_name) != -1:
                return True
        return False

    def get_command_result(self, command, timeout=None):
        tmp_code = locale.getdefaultlocale()[1]
        return self.get_command_result_with_encode(command, tmp_code, timeout)

    def get_command_result_with_encode(self, command, encode, timeout=None):
        if "|" in command:
            raise Exception("Special characters exist.")
        if self.auth_mode == str(AuthType.APP_PASSWORD.value):
            process = subprocess.Popen(shlex.split(command), stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                       stderr=subprocess.PIPE, encoding=encode, errors="ignore")
            user_pass = get_env_variable(self.pass_prefix)
            try:
                process.stdin.write(f"{user_pass}\n")
                process.stdin.flush()
                # 使用process.communicate()代替process.wait()
                out, err = process.communicate(timeout=timeout)
                code = str(process.returncode)
            finally:
                clear(user_pass)
            out = out.strip("Password:")
        else:
            process = subprocess.Popen(shlex.split(command), stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                       stderr=subprocess.PIPE, encoding=encode, errors="ignore")
            # 使用process.communicate()代替process.wait()
            out, err = process.communicate(timeout=timeout)
            code = str(process.returncode)
        return code, out, err

    # 获取SQL server版本
    def get_version(self):
        version = None
        cmd = f"sqlcmd {self.user_info} -Q \"SELECT SERVERPROPERTY('ProductVersion');\" " \
              f"-W -f {CodeIdConst.WINDOWS_CODE} -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        log.info(f'get original version, cmd: {cmd}, ret: {ret}, std_out: {std_out}')
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get original version for err: {std_err}, out: {std_out}")
            raise Exception("Failed to exec sql cmd")
        version = std_out.split('\n')[2].strip()
        return version

    def get_group_info(self, group_name):
        group_info_path = TmpFile.GROUP_INFO + self.job_id
        # 获取可用性组group_id
        group_info = dict()
        cmd = f"sqlcmd {self.user_info} -Q \"select group_id, resource_id from sys.availability_groups_cluster " \
              f"where name=N'{group_name}'\" -W -f {CodeIdConst.WINDOWS_CODE} " \
              f"-o \"{group_info_path}\" -l {CodeIdConst.TIMEOUT}"
        log.info(f"get_group_info cmd : [{cmd}]")
        code, std_out, std_err = self.get_command_result(cmd)
        if code != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get group info for err: {std_err}, out: {std_out}")
            raise Exception("Failed to exec sql cmd")
        tmp_out = self.read_tmp_result_file(group_info_path)
        json_dict = tmp_out.split("\n")[2]
        group_info['group_id'] = json_dict.split(" ")[0]
        group_info['resource_id'] = json_dict.split(" ")[1]
        return group_info

    def get_instance_databases(self):
        instance_list_path = TmpFile.INSTANCE_LIST_OUT + self.job_id
        cmd = f"sqlcmd {self.user_info} -Q \"SELECT name FROM sys.databases;\" -W -f 65001" \
              f" -o \"{instance_list_path}\" -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get databases for err: {std_err}, out: {std_out}")
            raise Exception("Failed to exec sql cmd")
        tmp_out = self.read_tmp_result_file(instance_list_path)
        database_list = tmp_out.split("\n")[2:-3]
        return database_list

    def get_instance_database_size_asc(self):
        instance_list_path = TmpFile.INSTANCE_LIST_OUT + self.job_id
        cmd = f"sqlcmd {self.user_info} -Q \"{SqlCmd.ORDER_DATABASES}\" -W -f 65001" \
              f" -o \"{instance_list_path}\" -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get databases for err: {std_err}, out: {std_out}")
            raise Exception("Failed to exec sql cmd")
        tmp_out = self.read_tmp_result_file(instance_list_path).split("\n")[2:-3]
        database_list = list()
        for info in tmp_out:
            total_size = info.split(" ")[-1].strip()
            database_name = info[:-(len(total_size) + 1)]
            database_list.append(database_name)
        return database_list

    def check_db_is_exist(self, database_list):
        instance_dbs = self.get_instance_databases()
        for database in database_list:
            if database in instance_dbs:
                log.error(f"Database: {database} exist")
                return True
        return False

    def get_availability_endpoint(self):
        endpoint_info_path = TmpFile.ENDPOINT_INFO + self.job_id
        # 获取endpoint
        cmd = f"sqlcmd {self.user_info} -Q \"SELECT port FROM sys.TCP_endpoints " \
              f"where type_desc='DATABASE_MIRRORING'\" -W -f {CodeIdConst.WINDOWS_CODE} " \
              f"-o \"{endpoint_info_path}\" -l {CodeIdConst.TIMEOUT}"
        code, std_out, std_err = self.get_command_result(cmd)
        if code != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get endpoint info as out: {std_out}, err: {std_err}")
            raise Exception("Failed to get endpoint info")
        tmp_out = self.read_tmp_result_file(endpoint_info_path)
        json_dict = tmp_out.split("\n")
        if len(json_dict) > 3:
            return json_dict[2]
        # 新集群可能获取不到，使用默认端口
        return SQLServerStrConstant.END_POINT

    def get_replica_info(self, group_name):
        replica_info_path = TmpFile.REPLICA_INFO + self.job_id
        # 获取可用行组副本信息
        group_info = self.get_group_info(group_name)
        if not group_info:
            raise Exception("Group info is empty")
        group_id = group_info.get("group_id")
        cmd = f"sqlcmd {self.user_info} -Q \"select replica_id, is_local, role from " \
              f"sys.dm_hadr_availability_replica_states where group_id='{group_id}'\" -W " \
              f"-f {CodeIdConst.WINDOWS_CODE} -o \"{replica_info_path}\" -l {CodeIdConst.TIMEOUT}"
        code, std_out, std_err = self.get_command_result(cmd)
        if code != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get replica info for err: {std_err}, out: {std_out}")
            raise Exception("Failed to exec sql cmd")
        tmp_out = self.read_tmp_result_file(replica_info_path)
        return tmp_out.split("\n")[2:-3]

    def get_always_on_database(self, group_name):
        ag_list_path = TmpFile.AG_LIST_OUT + self.job_id
        # 获取可用行组数据库
        replica_info = self.get_replica_info(group_name)
        replica_id = self.get_replica_id(replica_info)
        if not replica_id:
            log.error("Failed to get primary replica id")
            raise Exception("Failed to get primary replica id")
        cmd = f"sqlcmd {self.user_info} -Q \"select database_name from " \
              f"sys.dm_hadr_database_replica_cluster_states where replica_id='{replica_id}'\"" \
              f" -W -f 65001 -o \"{ag_list_path}\" -l {CodeIdConst.TIMEOUT}"
        code, std_out, std_err = self.get_command_result(cmd)
        if code != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get always on databases for std_err: {std_err}, out: {std_out}")
            raise Exception("Failed to exec sql cmd")
        database_list = list()
        tmp_out = self.read_tmp_result_file(ag_list_path).split("\n")
        if len(tmp_out) > 3:
            database_list = tmp_out[2:-3]
        return database_list

    def get_cluster_node_info(self, group_name):
        node_info_file_path = TmpFile.NODE_INFO_FILE + self.job_id
        group_info = self.get_group_info(group_name)
        if not group_info:
            raise Exception("Group info is empty")
        resource_id = group_info.get("resource_id")
        cmd = f"sqlcmd {self.user_info} -Q \"select instance_name, node_name from " \
              f"sys.dm_hadr_instance_node_map where ag_resource_id = '{resource_id}'\"" \
              f" -W -f 65001 -o \"{node_info_file_path}\" -l {CodeIdConst.TIMEOUT}"
        code, std_out, std_err = self.get_command_result(cmd)
        if code != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get cluster node info for err: {std_err}, out: {std_out}")
            raise Exception("Failed to exec sql cmd")
        tmp_out = self.read_tmp_result_file(node_info_file_path).split("\n")
        if len(tmp_out) <= 3:
            log.error(f"Failed to get cluster node info for info: {tmp_out}")
            raise Exception("Failed to exec sql cmd")
        cluster_info = list()
        for info in tmp_out[2:-3]:
            tmp_info = dict()
            tmp_info['instance_name'] = info.split(" ")[0]
            tmp_info['node_name'] = info.split(" ")[1]
            cluster_info.append(tmp_info)
        return cluster_info

    def check_is_primary_replica(self, group_name):
        # 检查是否是主副本节点
        replica_info = self.get_replica_info(group_name)
        for info in replica_info:
            is_local = info.split(" ")[1]
            role = info.split(" ")[2]
            log.info(f'get {group_name} role is {role}, is local is {is_local}')
            if is_local == "1" and role == "1":
                return True
        return False

    def check_version_is_matched(self, src_version):
        if not src_version:
            return True
        current_version = self.get_version()
        log.info(f"current_version:{current_version}, src_version:{src_version}")
        return current_version >= src_version

    def get_availability_replicas(self, group_name):
        ag_replica_info_path = TmpFile.AG_REPLICA_INFO + self.job_id
        group_info = self.get_group_info(group_name)
        if not group_info:
            raise Exception("Group info is empty")
        group_id = group_info.get("group_id")
        cmd = f"sqlcmd {self.user_info} -Q \"select endpoint_url, availability_mode_desc, " \
              f"failover_mode_desc from sys.availability_replicas where group_id='{group_id}'\" -W " \
              f"-f {CodeIdConst.WINDOWS_CODE} -o \"{ag_replica_info_path}\" -l {CodeIdConst.TIMEOUT}"
        code, std_out, std_err = self.get_command_result(cmd)
        if code != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get availability_replicas, err: {std_err}, out: {std_out}")
            raise Exception("Failed to exec sql cmd")
        tmp_out = self.read_tmp_result_file(ag_replica_info_path).split("\n")
        if len(tmp_out) <= 3:
            log.error(f"Failed to get availability_replicas, err {tmp_out}")
            raise Exception("Failed to get availability_replicas")
        replica_info = list()
        for info in tmp_out[2:-3]:
            tmp_info = dict()
            tmp_info["endpoint_url"] = info.split(" ")[0]
            tmp_info["availability_mode_desc"] = info.split(" ")[1]
            tmp_info["failover_mode_desc"] = info.split(" ")[2]
            replica_info.append(tmp_info)
        return replica_info

    def check_group_is_exist(self, group_name):
        out_file_path = TmpFile.OUT_FILE + self.job_id
        cmd = f"sqlcmd {self.user_info} -Q \"select name from sys.availability_groups_cluster\"" \
              f" -W -f 65001 -o \"{out_file_path}\" -l {CodeIdConst.TIMEOUT}"
        code, std_out, std_err = self.get_command_result(cmd)
        if code != ExecCmdResult.SUCCESS:
            raise Exception("Failed to get availability_groups_cluster info")
        tmp_out = self.read_tmp_result_file(out_file_path)
        group_info = tmp_out.split("\n")[2:-3]
        return group_name in group_info

    def get_database_file_info(self, database, flag):
        # 获取数据库本地文件信息，新位置恢复使用
        db_file_info_path = TmpFile.DB_FILE_INFO + self.job_id
        cmd = f"sqlcmd {self.user_info} -Q \"select physical_name, name from sys.master_files " \
              f"where database_id=db_id(N'{database}')\" -s ',' -W -f 65001 -o \"{db_file_info_path}\" " \
              f"-l {CodeIdConst.TIMEOUT}"
        ret, std_out, err = self.get_command_result(cmd)
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Failed to query the file info of database: {database}, out: {std_out}, err: {err}")
            raise Exception("Failed to query the file info")
        tmp_out = self.read_tmp_result_file(db_file_info_path).split("\n")[2:-3]
        file_info = list()
        for info in tmp_out:
            tmp_info = dict()
            name = info.split(",")[-1]
            tmp_info["name"] = name
            tmp_info["physical_name"] = info[:-(len(name) + 1)]
            file_info.append(tmp_info)
        if flag:
            for items in file_info:
                # 只保存文件名
                items["physical_name"] = os.path.basename(items.get("physical_name"))
        return file_info

    def get_restore_data_path(self):
        tmp_file_path = TmpFile.OUT_FILE + self.job_id
        cmd = f"sqlcmd {self.user_info} -Q \"SELECT serverproperty('InstanceDefaultDataPath') AS [DefaultDataPath]\"" \
              f" -W -f 65001 -o \"{tmp_file_path}\" -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        log.info(f"ret:{ret},std_out:{std_out},std_err:{std_err},cmd:{cmd}")
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get restore data path, out: {std_out}, err: {std_err}")
            raise Exception(f"Failed to get restore data path")
        tmp_out = self.read_tmp_result_file(tmp_file_path)
        if len(tmp_out.split("\n")) <= 3:
            return ""
        restore_data_path = tmp_out.split("\n")[2:-3]
        return restore_data_path[0]

    def get_ag_database_list(self):
        ag_database_path = TmpFile.AG_DATABASE_OUT + self.job_id
        cmd = f"sqlcmd {self.user_info} -Q \"select T3.name " \
              f"from sys.dm_hadr_database_replica_states T1 " \
              f"left JOIN sys.availability_groups T2 on T1.group_id = T2.group_id " \
              f"left JOIN master.sys.databases T3 on T1.database_id = T3.database_id\"" \
              f" -W -f 65001 -o \"{ag_database_path}\""
        ret, std_out, err = self.get_command_result(cmd)
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get ag database list as out: {std_out}, err: {err}")
            raise Exception("Failed to get ag database list")
        tmp_out = self.read_tmp_result_file(ag_database_path)
        if len(tmp_out.split("\n")) <= 3:
            return []
        database_list = tmp_out.split("\n")[2:-3]
        return database_list

    def check_database_in_use(self, database_list):
        dt_list = list()
        for database in database_list:
            cmd = f"sqlcmd {self.user_info} -Q \"SELECT * FROM [Master].[dbo].[SYSPROCESSES] " \
                  f"WHERE [DBID] IN (SELECT [DBID] FROM [Master].[dbo].[SYSDATABASES] " \
                  f"WHERE NAME='{database}') AND LASTWAITTYPE NOT LIKE '%REDO%'\"" \
                  f" -f {CodeIdConst.WINDOWS_CODE} -y \"0\" -l {CodeIdConst.TIMEOUT}"
            ret, std_out, err = self.get_command_result(cmd)
            log.info(f"ret:{ret},std_out:{std_out},std_err:{err}")
            if ret != ExecCmdResult.SUCCESS:
                log.error(f"Failed to check database: {database} in use or not, out: {std_out}, err: {err}")
                raise Exception(f"Failed to check database: {database} in use or not")
            if len(std_out.split("\n")) > 3:
                # 检查数据库数据库配置了内存优化数据相关的文件组
                result = self.check_in_memory_database(database, std_out)
                if result:
                    continue
                log.info(f"Database: {database} is in use")
                dt_list.append(database)
        return dt_list

    def get_simple_sql_list(self):
        """
        获取处于simple恢复模式的数据库列表，master, tempdb除外
        :return:
        """
        simple_list_path = TmpFile.SIMPLE_LIST_OUT + self.job_id
        sql_list = list()
        ignores_list = ["master", "tempdb"]
        cmd = f"sqlcmd {self.user_info} -Q \"SELECT name, recovery_model_desc FROM sys.databases\" " \
              f"-W -f 65001 -o \"{simple_list_path}\" -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        if ret != ExecCmdResult.SUCCESS:
            log.error(f'Exec find recover model failed, Err code: {ret}, Stand out: {std_out},'
                      f' Err msg: {std_err}')
            raise Exception('Exec find recover model failed')
        tmp_out = self.read_tmp_result_file(simple_list_path).split("\n")[2:-3]
        for mbr in tmp_out:
            mbr_name = mbr.split(" ")[0]
            mbr_model = mbr.split(" ")[1]
            if "SIMPLE" in mbr_model and mbr_name not in ignores_list:
                sql_list.append(mbr_name)
        return sql_list

    def get_abnormal_database_list(self):
        abnormal_list_path = TmpFile.ABNORMAL_LIST_OUT + self.job_id
        abnormal_list = list()
        cmd = f"sqlcmd {self.user_info} -Q \"SELECT name, state FROM sys.databases\"" \
              f" -W -f 65001 -o \"{abnormal_list_path}\" -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Exec cmd to get database state failed, out: {std_out}, err: {std_err}")
            raise Exception("Failed to get database state")
        tmp_out = self.read_tmp_result_file(abnormal_list_path).split("\n")[2:-3]
        for info in tmp_out:
            dt_state = info.split(" ")[-1]
            dt_name = info[:-(len(dt_state) + 1)]
            if dt_state != "0":
                abnormal_list.append(dt_name)
        return abnormal_list

    def get_database_info(self, database_name):
        db_state_info_path = TmpFile.DB_STATE_INFO + self.job_id
        cmd = f"sqlcmd {self.user_info} -Q \"select state,recovery_model_desc from sys.databases" \
              f" where name=N'{database_name}'\"" \
              f" -W -f 65001 -o \"{db_state_info_path}\" -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Exec cmd to get database state failed, out: {std_out}, err: {std_err}")
            raise Exception("Failed to get database state")
        tmp_out = self.read_tmp_result_file(db_state_info_path).split("\n")[2:-3]
        db_state_info = list()
        for info in tmp_out:
            db_state_info.append({
                "name": database_name,
                "state": info.split(" ")[0],
                "recovery_model_desc": info.split(" ")[1]
            })
        return db_state_info

    def drop_database(self, database):
        cmd = f"sqlcmd {self.user_info} -Q \"drop database [{database}]\" -f {CodeIdConst.WINDOWS_CODE}"
        code, std_out, std_err = self.get_command_result(cmd)
        if code == ExecCmdResult.SUCCESS:
            log.info(f"Succeed to delete database: {database}")
            return
        log.error(f"Failed to delete database: {database}, std_out: {std_out}, std_err: {std_err}")

    def set_database_offline(self, database):
        # 将数据库置于离线状态
        if database in SQLServerStrConstant.SYS_DATABASE:
            return True
        cmd = f"sqlcmd {self.user_info} -Q \"use master;alter database [{database}] " \
              f"set offline with rollback immediate;\" -f {CodeIdConst.WINDOWS_CODE}"
        ret, std_out, std_err = self.get_command_result(cmd)
        if ret == ExecCmdResult.SUCCESS:
            log.info(f"Succeed to set database: {database} offline")
            return True
        log.error(f"Failed to set database: {database} offline, std_out: {std_out}, std_err: {std_err}")
        return False

    def get_ag_node_info(self, resource_id):
        cmd = f"sqlcmd {self.user_info} -Q \"select node_name from " \
              f"sys.dm_hadr_instance_node_map where ag_resource_id = '{resource_id}'\"" \
              f" -W -f 65001 -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        if ret != ExecCmdResult.SUCCESS or len(std_out.split("\n")) <= 3:
            log.error(f"Failed to get node info for resource_id: {resource_id} for std_out: {std_out}")
            raise Exception("Failed to get node info for resource_id")
        return std_out.split("\n")[2:-3]

    def get_replica_server_name(self):
        cmd = f"sqlcmd {self.user_info} -Q \"SELECT replica_server_name from sys.availability_replicas;\"" \
              f" -W -f 65001 -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        log.info(f"cmd:{cmd},ret:{ret},std_out:{std_out},std_err:{std_err}")
        if ret != ExecCmdResult.SUCCESS or len(std_out.split("\n")) <= 3:
            log.error(f"Failed to get replica server name,std_out: {std_out},std_err:{std_err}")
            raise Exception("Failed to get replica server name")
        return std_out.split("\n")[2:-3]

    def get_sysadmin_user_list(self):
        sysadmin_list = list()
        tmp_file_path = TmpFile.OUT_FILE + self.job_id
        cmd = f"sqlcmd {self.user_info} -Q \"EXEC sp_helpsrvrolemember 'sysadmin';\" -o \"{tmp_file_path}\" " \
              f"-l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        if ret != ExecCmdResult.SUCCESS:
            return False, sysadmin_list
        try:
            result_list = self.read_tmp_result_file_by_line(tmp_file_path)
        except Exception as e_info:
            log.error(f"Get sysadmin user list failed, Msg: {e_info}")
            return False, sysadmin_list
        if "Msg" in result_list[0]:
            return False, sysadmin_list
        for usr_mbr in result_list[2:]:
            usr_name = usr_mbr[CodeIdConst.CONSTANT_129:]
            usrs = self.get_str_by_one_spcase(usr_name)
            sysadmin_list.append(usrs)
        return True, sysadmin_list

    def check_user_permission(self, auth_mode, user_name):
        flags, usr_list = self.get_sysadmin_user_list()
        if not flags:
            log.error("Get sysadmin user list failed")
            return False
        if str(auth_mode) == str(AuthType.OS_PASSWORD.value):
            return self.find_user_in_list(SqlServerDefaultUser.DEFAULTUSER, usr_list, False)
        elif str(auth_mode) == str(AuthType.APP_PASSWORD.value):
            return self.find_user_in_list(user_name, usr_list, True)
        else:
            return False

    def drop_database_from_always_on(self, group_name, database):
        cmd = f"sqlcmd {self.user_info} -Q \"use master;ALTER AVAILABILITY GROUP [{group_name}] " \
              f"REMOVE DATABASE [{database}];\""
        code, std_out, std_err = self.get_command_result(cmd)
        log.info(f"cmd:{cmd},code:{code},std_out:{std_out},std_err:{std_err}")
        if code != ExecCmdResult.SUCCESS:
            log.error(f"Failed to drop database: {database}")
            raise Exception(f"Failed to drop database: {database}, err: {std_err}")
        log.info(f"Succeed to drop database: {database}")

    def alter_database_readable_secondary(self, group_name):
        replica_server_list = self.get_replica_server_name()
        log.info(f"replica_server_list:{replica_server_list}")
        for server_name in replica_server_list:
            cmd = f"sqlcmd {self.user_info} -Q \"use master;ALTER AVAILABILITY GROUP [{group_name}] " \
                  f"MODIFY REPLICA ON '{server_name}' WITH (SECONDARY_ROLE (ALLOW_CONNECTIONS = ALL));\""
            code, std_out, std_err = self.get_command_result(cmd)
            log.info(f"code:{code},std_out:{std_out},std_err:{std_err}")

    def add_database_to_always_on(self, group_name, database):
        cmd = f"sqlcmd {self.user_info} -Q \"use master;ALTER AVAILABILITY GROUP [{group_name}] " \
              f"ADD DATABASE [{database}];\""
        _, std_out, _ = self.get_command_result(cmd)
        if f"Database {database}" in std_out:
            log.error(f"Failed to add database: {database} as out: {std_out}")
            raise Exception(f"Failed to add database")
        log.info(f"Succeed to add database: {database}")

    def secondary_database_join_always_on(self, group_name, database):
        cmd = f"sqlcmd {self.user_info} -Q \"use master;ALTER DATABASE [{database}] " \
              f"SET HADR AVAILABILITY GROUP = [{group_name}];\""
        _, std_out, _ = self.get_command_result(cmd)
        if f"Database {database}" in std_out:
            log.error(f"Failed to add secondary database: {database} as out: {std_out}")
            raise Exception(f"Failed to add secondary database")
        log.info(f"Succeed to add secondary database: {database}")

    def secondary_replica_add_always_on(self, group_name):
        cmd = f"sqlcmd {self.user_info} -Q \"ALTER AVAILABILITY GROUP [{group_name}] JOIN; \""
        _, std_out, _ = self.get_command_result(cmd)
        if "Failed to join" in std_out:
            log.error(f"Failed to add secondary replica as out: {std_out}")
            raise Exception(f"Failed to add secondary replica")
        log.info("Succeed to add secondary replica")

    def delete_availability_group(self, group_name):
        cmd = f"DROP AVAILABILITY GROUP [{group_name}];"
        code, std_out, std_err = self.get_command_result(f"sqlcmd {self.user_info} -Q \"{cmd}\"")
        if code == ExecCmdResult.SUCCESS:
            log.info(f"Succeed to delete availability group: {group_name}")
            return
        log.error(f"Failed to delete availability group: {group_name}, std_out: {std_out}, std_err: {std_err}")

    def get_latest_backup_meta_info(self, database, vdi_type):
        backup_type_dict = {
            1: "D",
            2: "I",
            3: "L"
        }
        backup_type = backup_type_dict.get(vdi_type)
        if database == "master":
            backup_type = "D"
        cmd = f"sqlcmd {self.user_info} -Q \"use msdb;select top 1 first_lsn,last_lsn,type," \
              f"DATEDIFF(s, '1970-01-01T00:00:00', backup_start_date) AS backup_start_date," \
              f"DATEDIFF(s, '1970-01-01T00:00:00', backup_finish_date) AS backup_finish_date," \
              f"database_name from msdb.dbo.backupset " \
              f"where database_name = '{database}' and type = '{backup_type}' order by backup_finish_date desc\" " \
              f" -W -f 65001 -l {CodeIdConst.TIMEOUT}"
        ret, std_out, std_err = self.get_command_result(cmd)
        log.info(f"ret:{ret},std_out:{std_out},std_err:{std_err}")
        if ret != ExecCmdResult.SUCCESS or len(std_out.split("\n")) <= 3:
            log.error(f"Failed to get backup meta info: {database} for std_out: {std_out}, std_error:{std_err}")
            raise Exception("Failed to get backup meta info")
        backup_meta_info = std_out.split("\n")[2:-3][1]
        meta_info_items = backup_meta_info.split(" ")
        if len(meta_info_items) < 5:
            log.error(f"Failed to query backup set: {meta_info_items}")
            raise Exception(f"Failed to query backup set")
        return {
            "first_lsn": meta_info_items[0],
            "last_lsn": meta_info_items[1]
        }

    @staticmethod
    def report_job_details(job_id: str, sub_job_details: dict):
        # 主动上报任务
        try:
            cur_time = str(int((time.time())))
            result_info = invoke_rpc_tool_interface(job_id + cur_time, "ReportJobDetails", sub_job_details)
        except Exception as err:
            log.error(f"Invoke rpc_tool interface exception, err: {err}.")
            return False
        if not result_info:
            return False
        ret_code = result_info.get("code", -1)
        if ret_code != int(CMDResult.SUCCESS):
            log.error(f"Invoke rpc_tool interface failed, result code: {ret_code}.")
            return False
        return True
    
    def check_in_memory_database(self, database, processes):
        version = self.get_version();
        main_version = version.split('.')[0]
        # 2014及以后版本支持In-Memory OLTP
        if int(main_version) < 12:
            return False
        cmd = f"sqlcmd {self.user_info} -Q \"use {database};SELECT name FROM sys.filegroups WHERE type = 'FX';\" \
         -f {CodeIdConst.WINDOWS_CODE} -y \"0\" -l {CodeIdConst.TIMEOUT}"
        ret, std_out, err = self.get_command_result(cmd)
        log.info(f"ret:{ret},std_out:{std_out},std_err:{err}")
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Failed to check in memory database: {database}, out: {std_out}, err: {err}")
            return False
        if len(std_out.split("\n")) > 4:
            log.info(f"The database: {database} is configured with filegroups related to memory-optimized data")
            if len(processes.split("\n")) == 4 and 'XTP_OFFLINE_CKPT' in processes.split("\n")[0]:
                return True
        return False

    def get_database_info_by_name(self, name):
        cmd = (f"sqlcmd {self.user_info} -Q \"SELECT database_id, compatibility_level, state FROM \
        master.sys.databases where name= N'{name}';\" -W -f {CodeIdConst.WINDOWS_CODE} -l {CodeIdConst.TIMEOUT}")
        ret, std_out, std_err = self.get_command_result_with_encode(cmd, 'UTF-8')
        log.info(f'get database info, cmd: {cmd}, ret: {ret}, std_out: {std_out}')
        if ret != ExecCmdResult.SUCCESS:
            log.error(f"Failed to get database info for err: {std_err}, out: {std_out}")
            raise Exception("Failed to exec sql cmd")
        info = std_out.split('\n')[2].strip()
        return info

    @staticmethod
    def get_report_service_database_info(instance_name):
        if instance_name.upper() == SQLServerStrConstant.SQLSERVER_DEFAULT_SERVICES.upper():
            report_list = ("ReportServer", "ReportServerTempDB")
        else:
            report_list = (f"ReportServer${instance_name}", f"ReportServer${instance_name}TempDB")
        return report_list
